/*
 * Copyright (C) 2012  - 2013 The playserver  Project
 * Author:longwuwei
 * All Rights Reserved.
 */

#include "Mediaservice.h"
Mediaservice::Mediaservice()
{
    HearBeatTick = 60;
    runFlag = 0;
    initFlag = 0;
}

Mediaservice::~Mediaservice()
{
    runFlag = 0;
}

int Mediaservice::Svr_Start(void)
{
    /*
     *
     * init svr config
     */
    if(this->svrConfig.init() != SUCCESS)
    {
        return FAILURE;
    }

    /**
     * init svr info class
     */
    this->svrInfo.init(&this->svrConfig,this->get_core_version());

    /**
     * init svr log
     */
    this->svrLog.init(this->svrConfig.log_path);
    this->svrLog.set_loglvl(this->svrConfig.loglevel);

    /**
	* init svr file mamange utils class
	*/
	if(this->svrFileManage.init(&this->svrConfig,&this->svrLog) != SUCCESS)
	{
		this->svrLog.log_fmt(LOG_ERR,"svrFileManage init error! \r\n");
		return FAILURE;
	}

    /*
     * init svr task
     */
    if(this->svrTask.init(&this->svrConfig) != SUCCESS)
    {
        this->svrLog.log_fmt(LOG_ERR,"svrTask init error! \r\n");
        return FAILURE;
    }

    /*
     * init svr client utils
     */
    if(this->svrClient.init(&this->svrConfig, &this->svrTask, &this->svrLog, &this->svrFileManage, DFLT_MIN_CONN_SIZ) != SUCCESS)
    {
        this->svrLog.log_fmt(LOG_ERR,"svrClient init error! \r\n");
        return FAILURE;
    }

    /*init data stream*/
	this->svrLiveTrans.set_stream_svr_type(this->svrConfig.svrtype);
	if(this->svrLiveTrans.init(this->svrConfig.max_channel,
		this->svrConfig.is_make_m3u8_list,
		&this->svrLog,
		this->svrConfig.is_save_clip,
		this->svrConfig.liverootpath) != SUCCESS)
	{
		return FAILURE;
	}


    /*
     * init svr net service
     */
    if(this->init(this->svrConfig.listen_ip, this->svrConfig.listen_port, svrConfig.svr_max_connect_count)  != SUCCESS)
    {
        this->svrLog.log_fmt(LOG_ERR,"service init error! \r\n");
        return FAILURE;
    }

    /*
     * svr run flag
     */
    runFlag = 1;

    /*
     * start sverver
     */
    return this->start();
}

int Mediaservice::Svr_Status(void)
{
    return this->runFlag;
}

int Mediaservice::Svr_Stop(void)
{
    if(runFlag == 1)
    {
        printf("NetServer stop...\r\n");
        runFlag = 0;
        return this->stop();
    }
    else
    {
        return 0;
    }
}

/*
* stream handle interface
*Svr_Read_Http_Live_Stream:read http live stream function
*/
NEXT_OPRATION  Mediaservice::Svr_Read_Http_Live_Stream(user_token * token,buf_chunk * sock_buf)
{
	NEXT_OPRATION nOpt = DO_CLOSE;
	if(this->svrLiveTrans.read_buf(token->id,sock_buf->buffer,sock_buf->buf_len,token->index,token->offset,token->trans_size) == SUCCESS)
	{
		if(token->trans_size == 0)
		{
			sock_buf->offset = 0;
			sock_buf->size = 0;
			nOpt = DO_WAIT;
		}
		else
		{
			sock_buf->offset = 0;
			sock_buf->size = (ulong)token->trans_size;
			nOpt = DO_SEND;
		}
	}
	return nOpt;
}

/*
* stream handle interface
*Svr_Read_Data_Stream:read data stream function
*/
NEXT_OPRATION  Mediaservice::Svr_Read_Data_Stream(user_token * token,buf_chunk * sock_buf)
{
	NEXT_OPRATION nOpt = DO_CLOSE;
	if(this->svrLiveTrans.read_chunk((uint8_t *)sock_buf->buffer,sock_buf->buf_len,token->id,token->index,token->offset,token->trans_size,token->recv_send_flag))
	{
		if(token->trans_size == 0)
		{
			sock_buf->offset = 0;
			sock_buf->size = 0;
			nOpt = DO_WAIT;
		}
		else
		{
			/*ok is not first read next.*/
			token->recv_send_flag = 200;
			sock_buf->offset = 0;
			sock_buf->size = (ulong)token->trans_size;
			nOpt = DO_SEND;
		}
	}
	return nOpt;
}
/*
* stream handle interface
*Svr_Write_Http_Live_Stream:write http live stream function
*/
NEXT_OPRATION  Mediaservice::Svr_Write_Http_Live_Stream(user_token * token,buf_chunk * sock_buf)
{
	NEXT_OPRATION nOpt = DO_CLOSE;

	if(token->index == 0)
	{
		token->index ++;
		token->tConnTime = time(NULL);
		uint64_t vdatalen = 0;
		int httpCode = 0;
		http::parse_response_header(sock_buf->buffer,sock_buf->size,httpCode,vdatalen);
		if(httpCode < 300 && httpCode >= 200)
		{
			if(vdatalen > 0)
			{
				this->svrTask.enable_stream_task(token->id);
				token->offset = sock_buf->size - vdatalen;
				if(SUCCESS != svrLiveTrans.write_buf(token->id,sock_buf->buffer + token->offset,(ulong)vdatalen))
				{
					return DO_CLOSE;
				}
			}
			return DO_RECV;
		}
		else
		{
			return DO_CLOSE;
		}
	}
	else if(SUCCESS == svrLiveTrans.write_buf(token->id,sock_buf->buffer,sock_buf->size))
	{
		if(token->index == 1)
		{
			this->svrTask.enable_stream_task(token->id);
		}
		token->index ++;
		if(time(NULL) - token->tConnTime > 5)
		{
			token->tConnTime = time(NULL);
			if(!this->svrTask.check_stream_task_is_exit(token->id))
			{
				return DO_CLOSE;
			}
		}
		return DO_RECV;
	}
	else
	{
		return DO_CLOSE;
	}

	return nOpt;
}

/*
* stream handle interface
*Svr_Write_Data_Stream:write data stream function
*/
NEXT_OPRATION  Mediaservice::Svr_Write_Data_Stream(user_token * token,buf_chunk * sock_buf)
{
	if(token->index == 0)
	{
		token->index ++;
		token->tConnTime = time(NULL);
		uint64_t vdatalen = 0;
		int httpCode = 0;
		http::parse_response_header(sock_buf->buffer,sock_buf->size,httpCode,vdatalen);
		if(httpCode < 300 && httpCode >= 200)
		{
			if(vdatalen > 0)
			{
				this->svrTask.enable_stream_task(token->id);
				token->offset = sock_buf->size - vdatalen;
				if(!svrLiveTrans.write_chunk(token->id,(uint8_t *)(sock_buf->buffer + token->offset),vdatalen))
				{
					return DO_CLOSE;
				}
			}
			return DO_RECV;
		}
		else
		{
			return DO_CLOSE;
		}
	}
	else if(svrLiveTrans.write_chunk(token->id,(uint8_t *)(sock_buf->buffer),sock_buf->size))
	{
		if(token->index == 1)
		{
			this->svrTask.enable_stream_task(token->id);
		}
		token->index ++;
		if(time(NULL) - token->tConnTime > 5)
		{
			token->tConnTime = time(NULL);
			if(!this->svrTask.check_stream_task_is_exit(token->id))
			{
				return DO_CLOSE;
			}
		}
		return DO_RECV;
	}
	else
	{
		return DO_CLOSE;
	}
}


/*
* net handle interface
*Svr_TransFile:trans file
*/
NEXT_OPRATION  Mediaservice::Svr_TransFile(user_token * token,buf_chunk * sock_buf)
{
    if(token->filefd >= 0)
    {
        /*
        * set total file trans size
        */
        if(token->trans_size == 0)
        {
            token->trans_size = token->hFile_size;
        }

        /*
        *set trans time out:3600 seconds
        */
        sock_buf->size = 0;
        sock_buf->offset = 0;
        token->have_trans_size = 0;
        token->recv_send_flag = FALSE;

        lseek(token->filefd,token->hFile_offset, SEEK_SET);
        fcntl(token->filefd, F_SETFD, FD_CLOEXEC);
        return DO_TRANSFILE;
    }

    return DO_CLOSE;
}

/*
*inherit virtual function
*/
NEXT_OPRATION Mediaservice::accept_completed(user_token *& token,buf_chunk * sock_buf)
{
    user_token * ptoken = NULL;
    svrClient.alloc(&ptoken);
    ptoken->type = INVALID_TYPE;
    token = ptoken;
    return DO_RECV;
}
/*
*connect_completed
*/
NEXT_OPRATION Mediaservice::connect_completed(user_token * token,buf_chunk * sock_buf)
{
    NEXT_OPRATION nOpt = DO_CLOSE;
    if(token != NULL)
    {
        switch(token->type)
        {
            case SVR_HEARTBEAT:
                {
                    sock_buf->offset = 0;
                    sock_buf->size = svrClient.struct_heartbeat_data(sock_buf->buffer,sock_buf->buf_len,
                    		this->svrClient.get_session_count(),
                    		this->get_connect_pool_use_count());

                    nOpt = DO_SEND;
                    this->svrLog.log_fmt(LOG_INFO,"connect ok,begin send:%s\r\n",sock_buf->buffer);
                }
                break;
            case CDN_AUTH_CLIENT:
				{
					sock_buf->offset = 0;
					sock_buf->size = this->svrClient.struct_auth_request_data(sock_buf->buffer,sock_buf->buf_len,token->key);
					nOpt = DO_SEND;
					this->svrLog.log_fmt(LOG_INFO,"connect ok,begin send:%s\r\n",sock_buf->buffer);
				}
            	break;
            case CAPTURE_UDPTS:
				{
					sock_buf->offset = 0;
					sock_buf->size = 0;
					nOpt = DO_RECV;
				}
				break;
			case CAPTURE_HTTP_STREAM:
				{
					sock_buf->offset = 0;
					sock_buf->size = svrClient.struct_stream_request_data(sock_buf->buffer,sock_buf->buf_len,token->id,token->fmt,token->key);
					nOpt = DO_SEND;
				}
				break;
			case CDN_LIVE_CLIENT:
				{
					uint64_t  index = 0;
					ulong     offset = 0;
					this->svrLiveTrans.get_current_data_stream_index(token->id,index,offset);
					this->svrLog.log_fmt(LOG_INFO,"steamid %u get stream index ,index %I64u,offset %u",token->id,index,offset);
					if(index > 0)
					{
						index ++;
					}
					sock_buf->offset = 0;
					sock_buf->size = svrClient.struct_stream_request_data(sock_buf->buffer,sock_buf->buf_len,token->id,token->fmt,token->key,index);
					nOpt = DO_SEND;
				}
				break;
            default:
              nOpt = DO_CLOSE;
              break;
        }
    }
    return nOpt;
}
/*
*receive_completed
*/
NEXT_OPRATION  Mediaservice::receive_completed(user_token * token,buf_chunk * sock_buf)
{
    if(token == NULL)
    {
        /*do none, must close*/
        return DO_CLOSE;
    }

    /*token type*/
    switch(token->type)
    {
		case CDN_LIVE_CLIENT:
			{
				sock_buf->size =  sock_buf->offset;
				sock_buf->offset = 0;
				return Svr_Write_Data_Stream(token,sock_buf);
			}
			break;
		case CAPTURE_HTTP_STREAM:
			{
				sock_buf->size =  sock_buf->offset;
				sock_buf->offset = 0;
				return Svr_Write_Http_Live_Stream(token,sock_buf);
			}
			break;
		case CAPTURE_UDPTS:
			{
				sock_buf->size =  sock_buf->offset;
				sock_buf->offset = 0;
				if(token->index == 0)
				{
					this->svrTask.enable_stream_task(token->id);
				}
				else
				{
					token->index ++;
				}
				if(SUCCESS == svrLiveTrans.write_buf(token->id,sock_buf->buffer,sock_buf->size))
				{
					return DO_RECV;
				}
				else
				{
					return DO_CLOSE;
				}
			}
			break;
        case SVR_HEARTBEAT:
        case CDN_AUTH_CLIENT:
			{
				sock_buf->size =  sock_buf->offset;
				sock_buf->offset = 0;
			    return DO_CLOSE;
			}
			break;
        case INVALID_TYPE:
            {
                token->index = 0;
                token->trans_size = 0;
                token->have_trans_size = 0;
                token->tConnTime = time(NULL);
                Client_Type ctype = svrClient.anylsis_client_type(sock_buf->buffer,sock_buf->offset,token);
                /*check is receive completed*/
                if(ctype == ON_RECV_TYPE)
                {
                    /*must < 32 KB*/
                    if(sock_buf->offset >= 32768)
                    {
                        return DO_CLOSE;
                    }
                    /*go on receive*/
                    return DO_RECV;
                }
                else
                {
                    this->svrLog.log_fmt(LOG_ALL,"[%s]Request Data:%s",token->key,sock_buf->buffer);
                    sock_buf->size =  sock_buf->offset;
                    sock_buf->offset = 0;
                }

                /*check request type*/
                if(ctype == PLAYER_CLIENT)
                {
                    return DO_WAIT;
                }
                else if(ctype == DOWNLOAD_CLIENT)
                {
                    return DO_WAIT;
                }
                else if(ctype == HTTP_HTB_CLIENT)
                {
                    /*send response*/
                    sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt);
                }
                else if(ctype == HTTP_ADMIN_CLIENT)
                {
                    /*send response*/
                    sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,HTML,this->svrInfo.readSvrInfo());
                }
                else if(ctype == HTTP_BROWSE)
                {
                    /*send response*/
                    sock_buf->size = (ulong)http::struct_crossxml_doma_response_header(sock_buf->buffer);
                }
                else if(ctype == CDN_LIVE_SVR
                		|| ctype == LIVE_STREMA_CLIENT
                		|| ctype == LIVE_TIMEBACK_CLIENT
                		|| ctype == LIVE_TIMEINFO_CLIENT
                		|| ctype == LIVE_RERUN_CLIENT
                		|| ctype == HTTP_M3U8_CLIENT
                		|| ctype == HTTP_M3U8TS_CLIENT)
                {
                	return DO_WAIT;
                }
                else
                {
                    /*set type INVALID ,return 400 code*/
                    token->type = INVALID_TYPE;
                    sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt,400);
                }
            }
            break;
        default:
        	return DO_CLOSE;
        	break;
    }

    /*default do send*/
    return DO_SEND;
}

/*
*send_completed
*/
NEXT_OPRATION Mediaservice::send_completed(user_token * token,buf_chunk * sock_buf)
{
    if(token == NULL)
    {
          /*do none, must close*/
          return DO_CLOSE;
    }

    switch(token->type)
    {
          case PLAYER_CLIENT:
          case DOWNLOAD_CLIENT:
          case LIVE_RERUN_CLIENT:
      	  case LIVE_TIMEBACK_CLIENT:
				{
				   return this->Svr_TransFile(token,sock_buf);
				}
				break;
      	  case HTTP_M3U8TS_CLIENT:
				{
					if(token->have_trans_size >= token->trans_size)
					{
						return DO_CLOSE;
					}
					uint64_t read_len = 0;
					ulong buf_len = sock_buf->buf_len;
					if(token->trans_size < token->have_trans_size + buf_len)
					{
						buf_len = (u_long)(token->trans_size - token->have_trans_size);
					}
					if(this->svrLiveTrans.read_buf(token->id,sock_buf->buffer,buf_len,token->index,token->offset,read_len) != SUCCESS)
					{
						return DO_CLOSE;
					}

					if(read_len == 0)
						return DO_CLOSE;

					sock_buf->size = (ulong)read_len;
					token->have_trans_size += read_len;
					return DO_SEND;
				}
				break;
          case HTTP_BROWSE:
          case HTTP_ADMIN_CLIENT:
          case HTTP_HTB_CLIENT:
          case LIVE_TIMEINFO_CLIENT:
      	  case HTTP_M3U8_CLIENT:
                  return DO_CLOSE;
                  break;
          case LIVE_STREMA_CLIENT:
				{
					token->have_trans_size += sock_buf->size;
					NEXT_OPRATION op = Svr_Read_Http_Live_Stream(token,sock_buf);
					if(op == DO_WAIT)
					{
						token->tConnTime = time(NULL);
					}
					return op;
				}
				break;
          case SVR_HEARTBEAT:
          case CDN_AUTH_CLIENT:
          case CAPTURE_HTTP_STREAM:
          case CDN_LIVE_CLIENT:
                {
                    token->recv_send_flag = FALSE;
                    sock_buf->offset = 0;
                }
                return DO_RECV;
          case CDN_LIVE_SVR:
				{
					token->have_trans_size += sock_buf->size;
					NEXT_OPRATION op = Svr_Read_Data_Stream(token,sock_buf);
					if(op == DO_WAIT)
					{
						token->tConnTime = time(NULL);
					}
					return op;
				}
				break;
          default:
                return DO_CLOSE;
                break;
    }
    /*default do recv*/
    return DO_RECV;
}

/**
 *trans_heartbeat,check session is alive
 */
NEXT_OPRATION  Mediaservice::trans_heartbeat(user_token * token)
{
	if(token == NULL)
	{
		  /*do none, must close*/
		  return DO_CLOSE;
	}

	/**
	 * check session status
	 * if not exit ,or auth FAILURE,both to close
	 */
	if(token->type == PLAYER_CLIENT)
	{
		if(AUTH_SUCCESS != this->svrClient.get_client_session_auth_status(token->key))
		{
			//token->have_trans_size
			return DO_CLOSE;
		}
	}

	/*
	 * go on trans file
	 * */
	return DO_TRANSFILE;
}

/*
*trans_completed
*/
NEXT_OPRATION  Mediaservice::trans_completed(user_token * token,buf_chunk * sock_buf)
{
	if(token == NULL)
	{
		/*do none, must close*/
		return DO_CLOSE;
	}
	/*
	*LIVE_RERUN_CLIENT and LIVE_TIMEBACK_CLIENT
	*muiltpart file
	*/
	if(token->type == LIVE_RERUN_CLIENT && token->privt != NULL)
	{
		token_ex * ptk = (token_ex *)token->privt;
		ptk->total_have_trans_size += token->have_trans_size;
		token->have_trans_size = 0;
		token->trans_size = 0;
		ptk->nb_file_index ++;
		if(ptk->nb_file_index < ptk->nb_file_num && ptk->total_have_trans_size < ptk->total_trans_size)
		{
			/*
			*set next trans
			*/
			token->filefd       = ptk->filelist[ptk->nb_file_index].hFile;
			token->hFile_size   = ptk->filelist[ptk->nb_file_index].hFile_size;
			token->hFile_offset = ptk->filelist[ptk->nb_file_index].hFile_offset;
			if(ptk->total_have_trans_size + ptk->filelist[ptk->nb_file_index].trans_size >  ptk->total_trans_size)
			{
				token->trans_size  = ptk->total_trans_size -  ptk->total_have_trans_size;
			}
			else
			{
				token->trans_size  = ptk->filelist[ptk->nb_file_index].trans_size;
			}

			/*
			*begin go on trans file
			*/
			token->recv_send_flag = TRUE;
			return Svr_TransFile(token,sock_buf);
		}
		else
		{
			this->svrLog.log_fmt(LOG_DEBUG,"[%s]rewath has trans total size:%I64u/%I64u",token->key,ptk->total_have_trans_size,ptk->total_trans_size);
		}
	}
	else if(token->type == LIVE_TIMEBACK_CLIENT && token->privt != NULL)
	{
		token_ex * ptk = (token_ex *)token->privt;
		ptk->total_have_trans_size += token->have_trans_size;
		token->have_trans_size = 0;
		token->trans_size = 0;
		/*close file*/
		if(token->filefd != INVALID_HANDLE_VALUE)
		{
			close(token->filefd);
			token->filefd  = INVALID_HANDLE_VALUE;
		}
		/*next file*/
		int open_flag = -1;
		token->filefd = this->svrLiveTrans.get_next_data_clip(token->id,token->fmt,ptk,this->svrConfig.liverootpath,token->hFile_offset,token->hFile_size,open_flag);
		if(open_flag == TRUE)
		{
			token->hFile_offset = 0;
			token->recv_send_flag = TRUE;
			return this->Svr_TransFile(token,sock_buf);
		}
		else if(open_flag == FALSE || open_flag < 0)
		{
			if(open_flag < 0)
			{
				this->svrLog.log_fmt(LOG_ERR,"[%s]LIVE_TIMEBACK_CLIENT open file error:%d!",token->key,open_flag);
			}
			/*check client session is success*/
			if(token->fmt == TS && SUCCESS != this->svrLiveTrans.init_ts_ctx(token->id,token->index,token->offset,token->trans_size))
			{
				return DO_CLOSE;
			}

			/*
			*change client type
			*/
			token->type = LIVE_STREMA_CLIENT;
			this->svrLog.log_fmt(LOG_INFO,"[%s]change LIVE_TIMEBACK_CLIENT to LIVE_STREMA_CLIENT,has trans:%I64u!",token->key,ptk->total_have_trans_size);
			return this->Svr_Read_Http_Live_Stream(token,sock_buf);
		}
	}
	/*default return close*/
    return DO_CLOSE;
}

/*
*send_wait_compledted
*/
NEXT_OPRATION Mediaservice::recv_or_send_wait_completed(user_token * token,buf_chunk * sock_buf)
{
    if(token == NULL)
    {
        return DO_CLOSE;
    }

    switch(token->type)
    {
        case PLAYER_CLIENT:
            {
                /*check wait is time out*/
                if(time(NULL) - token->tConnTime > 10)
                {
                     return DO_CLOSE;
                }

                /*
                *get session auth status
                *find client session,and check client session is success,but not HTTP_LIVE_NODE
                */
                AUTH_STATUS status = this->svrClient.get_client_session_auth_status(token->key);
                if(status == AUTH_FAILURE || status == AUTH_DISCONN_WAIT)
                {
                     return DO_CLOSE;
                }
                else if(status == AUTH_WAIT)
                {
                     return DO_WAIT;
                }

                /*
                *set first read flag = TRUE
                */
                token->recv_send_flag = TRUE;
                memset(sock_buf->buffer,'\0',sock_buf->buf_len);
                if(token->trans_size == 0)
                {
                     sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt,token->hFile_size);
                }
                else  if(token->fmt == FLV || token->fmt == FHV)
				{
					/*get video header*/
					ubyte * buf = new ubyte[4096];
					if(buf == NULL)
					{
						this->svrLog.log_fmt(LOG_INFO,"get video header,alloc memory error!");
						return DO_CLOSE;
					}
					int len = av_format::read_flv_header(token->filefd,buf,4096);
					if(len <= 0)
					{
						/*free buf*/
						delete buf;
						buf = NULL;

						this->svrLog.log_fmt(LOG_INFO,"get video header error,code:%d!",len);
						return DO_CLOSE;
					}
					/*struct header*/
					sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt,token->trans_size + len);
					memcpy(sock_buf->buffer + sock_buf->size,buf,len);
					sock_buf->size += len;

					/*free buf*/
					delete buf;
					buf = NULL;
				}
				else
				{
					 sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt,token->hFile_offset,token->hFile_offset + token->trans_size - 1,token->hFile_size);
				}

                /*send*/
                this->svrLog.log_fmt(LOG_ALL,"[%s]Response Data:%s",token->key,sock_buf->buffer);
                return DO_SEND;
                /*end*/
            }
            break;
        case LIVE_STREMA_CLIENT:
			{
				/*check wiat is time out*/
				if(time(NULL) - token->tConnTime > 10)
				{
					return DO_CLOSE;
				}
				/*
				*get session auth status
				*find client session,and check client session is success,but not HTTP_LIVE_NODE
				*/
				AUTH_STATUS status = this->svrClient.get_client_session_auth_status(token->key);
				if(status == AUTH_FAILURE || status == AUTH_DISCONN_WAIT)
				{
					 return DO_CLOSE;
				}
				else if(status == AUTH_WAIT)
				{
					 return DO_WAIT;
				}

				/*read data check*/
				if(token->recv_send_flag == FALSE)
				{
					token->recv_send_flag = TRUE;

					/*get stream id*/
					if(token->privt != NULL)
					{
						token_ex * ptk = (token_ex *)token->privt;
						token->id = this->svrTask.find_stream_id(ptk->stream_name);
					}

					/*check client session is success*/
					if(token->fmt == TS && SUCCESS != this->svrLiveTrans.init_ts_ctx(token->id,token->index,token->offset,token->trans_size))
					{
						return DO_CLOSE;
					}
					memset(sock_buf->buffer,'\0',sock_buf->buf_len);
					sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt);
					sock_buf->offset = 0;
					return DO_SEND;
				}
				else
				{
					return Svr_Read_Http_Live_Stream(token,sock_buf);
				}
				/*end*/
			}
			break;
        case LIVE_RERUN_CLIENT:
			{
				/*check wiat is time out*/
				if(time(NULL) - token->tConnTime > 10)
				{
					return DO_CLOSE;
				}

				/*
				*get session auth status
				*find client session,and check client session is success,but not HTTP_LIVE_NODE
				*/
				AUTH_STATUS status = this->svrClient.get_client_session_auth_status(token->key);
				if(status == AUTH_FAILURE || status == AUTH_DISCONN_WAIT)
				{
					return DO_CLOSE;
				}
				else if(status == AUTH_WAIT)
				{
					return DO_WAIT;
				}


				/*
				*log start
				*/
				this->svrLog.log_fmt(LOG_DEBUG,"[%s]rewath request:%s",token->key,sock_buf->buffer);

				/*open file*/
				bool checkfile = false;
				token_ex * ptk = NULL;
				if(token->privt != NULL)
				{
					ptk = (token_ex *)token->privt;
					token->id = this->svrTask.find_stream_id(ptk->stream_name);
					/*
					*recal live watch to vod
					*/
					if(ptk->is_init)
					{
						checkfile = DataClip::open_clip_file_list(token->id,token->fmt,ptk,this->svrConfig.liverootpath);
					}
				}
				if(!checkfile)
				{
					this->svrLog.log_fmt(LOG_ERR,"[%s]rewath open request file error!",token->key);
					return DO_CLOSE;
				}
				else
				{
					this->svrLog.log_fmt(LOG_ERR,"[%s]rewath request file list begin count:%d ==",token->key,ptk->nb_file_num);
					for(int j= 0;j < ptk->nb_file_num ;j++)
					{
						this->svrLog.log_fmt(LOG_DEBUG,"[%s] [%s] size:%I64u,offset:%I64u,trans size:%I64u",
							token->key,
							ptk->filelist[j].hFileName,
							ptk->filelist[j].hFile_size,
							ptk->filelist[j].hFile_offset,
							ptk->filelist[j].trans_size);
					}
					this->svrLog.log_fmt(LOG_ERR,"[%s]rewath request file list end  ==",token->key);
				}

				/*
				*struct response header
				*/
				token->recv_send_flag = TRUE;
				memset(sock_buf->buffer,'\0',sock_buf->buf_len);
				if(!ptk->is_rang)
				{
					ptk->nb_file_index = 0;
					ptk->total_trans_size = ptk->total_file_size;
					ptk->total_have_trans_size = 0;

					token->filefd        = ptk->filelist[0].hFile;
					token->hFile_size   = ptk->filelist[0].hFile_size;
					token->hFile_offset = ptk->filelist[0].hFile_offset;
					token->trans_size   = ptk->filelist[0].trans_size;

					/*
					*buf struct
					*/
					sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt,ptk->total_file_size);
				}
				else
				{
					/*
					*check seek rang,and set total trans size
					*/
					if(ptk->total_offset >= ptk->total_file_size)
					{
						ptk->total_offset = (ptk->total_file_size - 1);
					}
					if(ptk->total_offset_end == 0 || ptk->total_offset_end  >= ptk->total_file_size)
					{
						ptk->total_offset_end = (ptk->total_file_size -1);
					}
					ptk->total_trans_size = (ptk->total_offset_end - ptk->total_offset + 1);
					ptk->total_have_trans_size = 0;

					/*
					*set begin clip,and end clip,
					*other clip after end clip must close ?
					*/
					uint64_t seek_offset = 0;
					bool begin_seekflag = false,end_seekflag = false;
					for(int i= 0;i < ptk->nb_file_num ;i++)
					{
						if(seek_offset +  ptk->filelist[i].trans_size > ptk->total_offset && !begin_seekflag)
						{
							ptk->nb_file_index  = i;
							token->filefd        = ptk->filelist[ptk->nb_file_index].hFile;
							token->hFile_size   = ptk->filelist[ptk->nb_file_index].hFile_size;
							token->hFile_offset = ptk->filelist[ptk->nb_file_index].hFile_offset + (ptk->total_offset - seek_offset);
							token->trans_size   = ptk->filelist[ptk->nb_file_index].trans_size - (ptk->total_offset - seek_offset);
							begin_seekflag = true;
						}

						if(seek_offset +  ptk->filelist[i].trans_size > ptk->total_offset_end && !end_seekflag)
						{
							ptk->filelist[i].trans_size = ptk->total_offset_end - seek_offset + 1;
							end_seekflag = true;
						}

						/*next*/
						if(begin_seekflag && end_seekflag)
							break;

						seek_offset += ptk->filelist[i].trans_size;
					}

					/*
					*buf struct
					*/
					sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt,ptk->total_offset,ptk->total_offset + ptk->total_trans_size - 1,ptk->total_file_size);
				}

				/*
				*log end
				*/
				this->svrLog.log_fmt(LOG_DEBUG,"[%s]rewath response:%s",token->key,sock_buf->buffer);

				return DO_SEND;
			}
			break;
		case LIVE_TIMEBACK_CLIENT:
			{
				/*check wiat is time out*/
				if(time(NULL) - token->tConnTime > 10)
				{
					return DO_CLOSE;
				}
				/*
				*get session auth status
				*find client session,and check client session is success,but not HTTP_LIVE_NODE
				*/
				AUTH_STATUS status = this->svrClient.get_client_session_auth_status(token->key);
				if(status == AUTH_FAILURE || status == AUTH_DISCONN_WAIT)
				{
					return DO_CLOSE;
				}
				else if(status == AUTH_WAIT)
				{
					return DO_WAIT;
				}


				/*add your code here,to set first file*/
				if(token->privt != NULL)
				{
					token_ex * ptk = (token_ex *)token->privt;
					token->id = this->svrTask.find_stream_id(ptk->stream_name);

					int open_flag = -1;
					token->filefd = this->svrLiveTrans.get_next_data_clip(token->id,token->fmt,ptk,this->svrConfig.liverootpath,token->hFile_offset,token->hFile_size,open_flag);
					if(open_flag == TRUE)
					{
						token->trans_size = token->hFile_size - token->hFile_offset;
					}
					else if(open_flag == FALSE || open_flag < 0)
					{
						if(open_flag < 0)
						{
							this->svrLog.log_fmt(LOG_ERR,"[%s]LIVE_TIMEBACK_CLIENT open file error:%d!",token->key,open_flag);
						}
						/*check client session is success*/
						if(token->fmt == TS && SUCCESS != this->svrLiveTrans.init_ts_ctx(token->id,token->index,token->offset,token->trans_size))
						{
							return DO_CLOSE;
						}

						/*
						*change client type
						*/
						token->type = LIVE_STREMA_CLIENT;
						this->svrLog.log_fmt(LOG_INFO,"[%s]change LIVE_TIMEBACK_CLIENT to LIVE_STREMA_CLIENT!",token->key);
					}
				}
				/*struct response header*/
				memset(sock_buf->buffer,'\0',sock_buf->buf_len);
				sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt);
				sock_buf->offset = 0;
				return DO_SEND;
			}
			break;
        case DOWNLOAD_CLIENT:
            {
                token->recv_send_flag = TRUE;
                sock_buf->offset = 0;
                memset(sock_buf->buffer,'\0',sock_buf->buf_len);
                if(token->trans_size == 0)
                {
                    sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt,token->hFile_size);
                }
                else
                {
                    sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt,token->hFile_offset,token->hFile_offset + token->trans_size - 1,token->hFile_size);
                }
                this->svrLog.log_fmt(LOG_ALL,"[%s]Response Data:%s",token->key,sock_buf->buffer);
                return DO_SEND;
            }
            break;
        case CDN_LIVE_SVR:
			{
				/*check wiat is time out*/
				if(time(NULL) - token->tConnTime > 10)
				{
					return DO_CLOSE;
				}
				/*read data check*/
				if(token->recv_send_flag == 0)
				{
					/*
					*set first read flag = TRUE
					*/
					token->recv_send_flag = TRUE;
					memset(sock_buf->buffer,'\0',sock_buf->buf_len);
					sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt);
					sock_buf->offset = 0;
					return DO_SEND;
				}
				else
				{
					return Svr_Read_Data_Stream(token,sock_buf);
				}
				/*end*/
			}
			break;
        case LIVE_TIMEINFO_CLIENT:
			{
				/*
				*get live stream
				*current time info
				*/
				if(token->privt != NULL)
				{
					token_ex * ptk = (token_ex *)token->privt;
					token->id = this->svrTask.find_stream_id(ptk->stream_name);
					this->svrLiveTrans.get_current_data_stream_timeinfo(token->id,ptk);
					char timeinfo[32];
					memset(timeinfo,'\0',sizeof(timeinfo));
					sprintf(timeinfo,"%04d-%02d-%02d %02d:%02d:%02d",ptk->year,ptk->month,ptk->day,ptk->hours,ptk->minit,ptk->seconds);
					/*
					*struct response header and data
					*/
					token->recv_send_flag = TRUE;
					memset(sock_buf->buffer,'\0',sock_buf->buf_len);
					sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,HTML,timeinfo);
					return DO_SEND;
				}
				else
				{
					return DO_CLOSE;
				}
			}
			break;
        case HTTP_M3U8_CLIENT:
			{
				/*check wiat is time out*/
				if(time(NULL) - token->tConnTime > 10)
				{
					return DO_CLOSE;
				}
				/*
				*get session auth status
				*find client session,and check client session is success,but not HTTP_LIVE_NODE
				*/
				AUTH_STATUS status = this->svrClient.get_client_session_auth_status(token->key);
				if(status == AUTH_FAILURE || status == AUTH_DISCONN_WAIT)
				{
					 return DO_CLOSE;
				}
				else if(status == AUTH_WAIT)
				{
					 return DO_WAIT;
				}

				/*get stream id*/
				if(token->privt != NULL)
				{
					token_ex * ptk = (token_ex *)token->privt;
					token->id = this->svrTask.find_stream_id(ptk->stream_name);
				}

				/*read m3u8 list check*/
				char m3u8list[4096];/*4KB*/
				memset(m3u8list,'\0',4096);
				size_t list_len = this->svrLiveTrans.read_m3u8_list(token->id,m3u8list,4096,token->key);
				if(list_len == 0)
				{
					return DO_CLOSE;
				}
				memset(sock_buf->buffer,'\0',sock_buf->buf_len);
				sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt,m3u8list);
				return DO_SEND;
				/*end*/
			}
			break;
		case HTTP_M3U8TS_CLIENT:
			{
				/*find client session,and check client session is success,but not HTTP_LIVE_NODE*/
				if(this->svrClient.get_client_session_auth_status(token->key) != AUTH_SUCCESS)
				{
					return DO_CLOSE;
				}
				/*init m3u8 ts clip*/
				this->svrLiveTrans.init_ts_ctx(token->id,token->index,token->offset,token->trans_size);
				if(token->trans_size == 0)
				{
					return DO_CLOSE;
				}
				memset(sock_buf->buffer,'\0',sock_buf->buf_len);
				sock_buf->size = (ulong)http::struct_response_header(sock_buf->buffer,(HTTP_MIME)token->fmt,token->trans_size);
				return DO_SEND;
				/*end*/
			}
			break;
        default:
            return DO_CLOSE;
            break;
    }

    return DO_WAIT;
}

/*
*send_wait_compledted
*/
void Mediaservice::close_completed(user_token * token,buf_chunk * sock_buf,NET_ERROR_CODE error)
{
    /*connect close log*/
    if(error != INVALID_ERROR && error != CLIENT_CLOSED)
    {
        this->svrLog.log_fmt(LOG_INFO,"client close completed,code:%d",error);
    }
    /*handle close opration*/
    if(token != NULL)
    {
        switch(token->type)
        {
            case PLAYER_CLIENT:
            case LIVE_RERUN_CLIENT:
			case LIVE_STREMA_CLIENT:
			case LIVE_TIMEBACK_CLIENT:
			case HTTP_M3U8_CLIENT:
			case HTTP_M3U8TS_CLIENT:
                {
                	 this->svrClient.close_client_session(token->key,FALSE,token->have_trans_size);
                }
                break;
            case SVR_HEARTBEAT:
                {
                	/*handle heartbeat data*/
					this->svrTask.handle_heartbeat_data(sock_buf->buffer,sock_buf->size);
					this->svrLog.log_fmt(LOG_INFO,sock_buf->buffer);
                    svrTask.svr_heartbeat_task = TASK_CLOSE;
                }
                break;
            case CDN_AUTH_CLIENT:
				{
					/**
					 * must go to anylsis auth result
					 * and remove client session
					 */
					this->svrClient.anylsis_auth_result(sock_buf->buffer,sock_buf->size,token->key);
				}
				break;
            case CAPTURE_HTTP_STREAM:
			case CAPTURE_UDPTS:
				{
					svrLiveTrans.free_stream(token->id);
					int retval = svrTask.close_stream_task(token->id);
					if(retval != 0)
					{
						/*other handle*/
					}
					this->svrLog.log_fmt(LOG_INFO,"task end ,taskid:[%d],media_fmt:[%d],stream_name:%s,type:%d,return:%d",token->id,token->fmt,token->key,token->type,retval);
				}
				break;
			case CDN_LIVE_CLIENT:
				{
					svrLiveTrans.close_data_stream(token->id,FALSE);
					int retval = svrTask.close_stream_task(token->id);
					if(retval != 0)
					{
						svrLiveTrans.close_data_stream(token->id,TRUE);
					}
					this->svrLog.log_fmt(LOG_INFO,"task end ,taskid:[%d],media_fmt:[%d],stream_name:%s,type:%d,return: %d",token->id,token->fmt,token->key,token->type,retval);
				}
				break;
            default:break;
        }

        /*
        *must free token
        */
        svrClient.free(token);
    }
}

/*
*svr log
*/
int  Mediaservice::log_fmt(int nloglvl,const char * lpszFormat , ...)
{
    va_list argList;
    va_start(argList, lpszFormat);

    char m_szMsg[8192];
    memset(m_szMsg,'\0',8192);
    vsprintf(m_szMsg, lpszFormat, argList);
    this->svrLog.log_fmt_arglist(nloglvl,lpszFormat,argList);

    va_end(argList);

    return SUCCESS;
}

void  Mediaservice::Svr_HearBeat(void)
{
	/*update config*/
	svrConfig.update();

	/*update stream config*/
	svrTask.init_config_stream();

    /*HearBeat*/
    if(svrTask.svr_heartbeat_task == TASK_CLOSE)
    {
        /*update config*/
        this->svrConfig.update();
        this->svrConfig.set_authsvr_address();

        /*get stream post connect*/
         svrTask.svr_heartbeat_task = TASK_RUN;

         user_token * token = NULL;
         svrClient.alloc(&token);
         token->type = SVR_HEARTBEAT;
         token->local_addr.sin_family = AF_INET;
         token->local_addr.sin_addr.s_addr = htonl(INADDR_ANY);
         token->local_addr.sin_port = htons(0);
         token->remote_addr.sin_family = AF_INET;
         token->remote_addr.sin_addr.s_addr = inet_addr(this->svrConfig.auth_ip);
         token->remote_addr.sin_port = htons(this->svrConfig.auth_port);
         post_usr_connect(token,NET_TCP);
    }
}

void  Mediaservice::Svr_Run_task(void)
{
	/**
	 * run task list
	 * */
	live_stream_task * ptask = this->svrTask.begin_enter();
	{
		while(ptask != NULL)
		{
			this->svrLog.log_fmt(LOG_INFO,"task begin ,taskid:[%d],media_fmt:[%d],stream_name:%s",ptask->id,ptask->stream_fmt,ptask->stream_name);

			if(ptask->status == TASK_CLOSE)
			{
				ptask->status = TASK_RUN;
				ptask->begin_time = time(NULL);

				/*
				*check task type
				*/
				if(ptask->stream_fmt == FLV || ptask->stream_fmt == TS)
				{
					user_token * token = NULL;
					svrClient.alloc(&token);
					token->local_addr.sin_family = AF_INET;
					token->remote_addr.sin_family = AF_INET;
					/*bind local*/
					token->local_addr.sin_addr.s_addr = htonl(INADDR_ANY);
					token->local_addr.sin_port = htons(0);
					/*set remote*/
					token->remote_addr.sin_addr.s_addr = inet_addr(ptask->remote_ip);
					token->remote_addr.sin_port = htons(ptask->addr_port);

					token->id = ptask->id;
					token->fmt = ptask->stream_fmt;
					strcpy(token->key,ptask->stream_name);
					if(this->svrConfig.svrtype == 1)
					{
						token->type = CAPTURE_HTTP_STREAM;
						svrLiveTrans.init_stream(token->id,ptask->stream_fmt);
					}
					else
					{
						token->type = CDN_LIVE_CLIENT;
						svrLiveTrans.init_data_stream(token->id,ptask->stream_fmt,NULL,0);
					}
					post_usr_connect(token,NET_TCP);
				}
				else if(ptask->stream_fmt == UDPTS)
				{
					user_token * token = NULL;
					svrClient.alloc(&token);
					token->local_addr.sin_family = AF_INET;
					token->remote_addr.sin_family = AF_INET;
					/*bind local*/
					token->local_addr.sin_addr.s_addr = inet_addr(ptask->remote_ip);
					token->local_addr.sin_port = htons(ptask->addr_port);

					token->id = ptask->id;
					token->fmt = TS;
					strcpy(token->key,ptask->stream_name);
					token->type = CAPTURE_UDPTS;

					/*
					*init stream
					*notice:init stream do not check return value
					*/
					this->svrLiveTrans.init_stream(token->id,TS);
					post_usr_connect(token,NET_UDP);
				}
			}
			/*
			 * next task
			 */
			ptask = ptask->next;
		}

		/**run end**/
		this->svrTask.end_enter();
	}
}

void  Mediaservice::Svr_Handle_Client_Session(void)
{
	/*auth*/
	int i = 0;
	for(i = 0;i < 100 ;i++)
	{
		user_token * token = this->svrClient.get_next_session_auth_task();
		if(token == NULL)
		{
			break;
		}
		else
		{
			this->svrLog.log_save_db(LOG_INFO,"[%s][%d] Auth begin,authip: %s:%d",token->key,token->type,this->svrConfig.auth_ip,this->svrConfig.auth_port);
			token->local_addr.sin_family = AF_INET;
			token->local_addr.sin_addr.s_addr = htonl(INADDR_ANY);
			token->local_addr.sin_port = htons(0);
			token->remote_addr.sin_family = AF_INET;
			token->remote_addr.sin_addr.s_addr = inet_addr(this->svrConfig.auth_ip);
			token->remote_addr.sin_port = htons(this->svrConfig.auth_port);
			post_usr_connect(token,NET_TCP);
		}
	}
}

int Mediaservice::Svr_Run(long useconds)
{
    if(initFlag == 0)
    {
        initFlag = 1;
    }

    /*
     * svr heartbeat
     */
    if(this->HearBeatTick * useconds >= 10000000)
    {
        Svr_HearBeat();

        this->HearBeatTick = 0;
    }

    /*
     * run task
     */
    this->Svr_Run_task();

    /*
     * handle client session
     */
    this->Svr_Handle_Client_Session();

    /*
	 * svrClient handle run
	 * */
	this->svrClient.run(useconds);


	/**
	 * svr file manage run
	 */
	this->svrFileManage.run(useconds);

	/**
	 *refresh  server info
	 */
	this->svrInfo.refresh(useconds,this->get_connect_pool_use_count(),
			this->get_connect_pool_unuse_count(),
			this->get_connect_pool_count(),
			this->get_buf_pool_unuse_count(),
			this->get_buf_pool_count(),
			this->svrClient.get_free_count(),
			this->svrClient.get_count(),
			this->svrClient.get_session_count(),
			this->svrClient.get_free_session_count(),
			this->svrLiveTrans.get_current_stream_count(),
			this->svrLiveTrans.get_free_stream_count(),
			this->svrLiveTrans.get_current_packet_count(),
			this->svrLiveTrans.get_free_packet_count(),
			this->svrLiveTrans.get_current_chunk_count(),
			this->svrLiveTrans.get_free_chunk_count(),
			this->svrLiveTrans.get_current_data_stream_count(),
			this->svrLiveTrans.get_free_data_stream_count());

		/*this->svrLiveTrans.get_current_stream_count(),
		this->svrLiveTrans.get_free_stream_count(),
		this->svrLiveTrans.get_current_packet_count(),
		this->svrLiveTrans.get_free_packet_count(),
		this->svrLiveTrans.get_current_chunk_count(),
		this->svrLiveTrans.get_free_chunk_count(),
		this->svrLiveTrans.get_current_data_stream_count(),
		this->svrLiveTrans.get_free_data_stream_count()*/

    /*
     *time tick
     */
    this->HearBeatTick ++;

    return 0;
}
